/* Copyright (C) 2008  Romolo Camplani, Marco Fortunato, Marco Marelli, Guido Rota, Fabio A. Schreiber et al.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */
package org.dei.perla.sys.device.channel;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.dei.perla.component.Component;
import org.dei.perla.sys.device.channel.frame.Frame;
import org.dei.perla.sys.device.channel.frame.FrameHeaderFul;
import org.dei.perla.sys.device.channel.frame.FrameHeaderLess;
import org.dei.perla.sys.device.channel.frame.address.FrameAddress;
import org.dei.perla.sys.device.channel.frame.header.FrameHeader;
import org.dei.perla.sys.device.channel.frame.receiver.FrameReceiver;
import org.dei.perla.sys.device.channel.frame.receiver.FrameReceiverEvents;
import org.dei.perla.sys.device.channel.frame.receiver.FrameReceiverHeaderFul;
import org.dei.perla.sys.device.channel.frame.receiver.FrameReceiverHeaderLess;
import org.dei.perla.sys.device.channel.frame.sender.FrameSender;
import org.dei.perla.sys.device.channel.frame.sender.FrameSenderEvents;
import org.dei.perla.sys.device.channel.frame.sender.FrameSenderHeaderFul;
import org.dei.perla.sys.device.channel.frame.sender.FrameSenderHeaderLess;
import org.dei.perla.utils.logger.LogRecord.Type;
import org.dei.perla.utils.logger.LogRecord.Verbosity;
import org.dei.perla.utils.messages.AdapterChannelMessage;
import org.dei.perla.utils.pipe.Pipe;
import org.dei.perla.utils.waiter.Waitable;
import org.dei.perla.utils.waiter.Waiter;


/**
 * Questa classe modella un canale generico.
 * Un canale può supportare un solo tipo di frame (headerless/headerful).
 * Un canale è bidirezionale e sempre instaurato tra due nodi.
 */
public abstract class Channel extends Component {
		
	/**
	 * Oggetto che implementa l'automa di ricezione dei frame
	 */
	private FrameReceiver pFrameReceiver;
	
	/**
	 *  Oggetto per ricevere gli eventi provenienti dall'automa di ricezione dei frame
	 */
	private FrameReceiverEvents pFrameReceiverEvents;
	
	/**
	 * Oggetto che implementa il meccanismo di costruzione dei frame
	 */
	private FrameSender pFrameSender;
	
	/**
	 *  Oggetto per ricevere gli eventi provenienti dal FrameSender
	 */
	private FrameSenderEvents pFrameSenderEvents;
	
	/**
	 * Input Stream per leggere dal canale
	 */
	private OutputStream pChannelOutputStream;
	
	/**
	 * Output stream per scrivere sul canale
	 */
	private InputStream pChannelInputStream;	
	
	/**
	 * Pipe usata per ricevere i frame da inviare sul canale
	 */
	private Pipe<AdapterChannelMessage> pOutputPipe;
	
	/**
	 * Pipe usata per notificare i frame ricevuti dal canale
	 */
	private Pipe<AdapterChannelMessage> pInputPipe;
	
	/**
	 * Waiter usato per attendere la presenza di dati nella Pipe di output
	 */
	private Waiter pOutputPipeWaiter;
	
	/**
	 * Indica se l'input dal canale sarà gestito ad eventi o da un thread apposito.
	 * Nel primo caso sarà cioè il thread del driver a chiamare, quando ci sono dati da ricevere, 
	 * la funzione pReceive. Nel secondo caso invece sarà avviato un thread che effettua delle
	 * letture bloccanti sul canale di input.
	 */
	private boolean pEventBasedInput;
	
	/**
	 * Thread che gestisce le operazioni di invio sul canale
	 */
	private Thread pThreadSender;	
	
	/**
	 * Thread che gestisce le operazioni di ricezione dal canale
	 */
	private Thread pThreadReceiver;
	
	/**
	 * Indirizzo locale della macchina
	 */
	protected FrameAddress pLocalAddress;
	
	/**
	 * Indirizzo remoto all'altro capo del canale
	 */
	protected FrameAddress pRemoteAddress;
	
	/**
	 * Costruttore privato. Inizializza il canale per frame sia headerless/full.
	 * @param parName Nome del canale
	 * @param parFrameReceiver Ricevitore dei frames
	 * @param parFrameSender Oggetto per inviare i frames
	 * @param parLocalAddress Indirizzo locale del canale
	 */
	private Channel(String parName, FrameReceiver parFrameReceiver, FrameSender parFrameSender, FrameAddress parLocalAddress) {
		
		super(parName);
		this.pFrameReceiver = parFrameReceiver;
		this.pFrameSender 	= parFrameSender;
		
		// Istanzia le Pipe necessarie per comunicare
		this.pOutputPipe 	= new Pipe<AdapterChannelMessage>(parName + "_snd");
		this.pInputPipe 	= new Pipe<AdapterChannelMessage>(parName + "_rcv");
		
		// Istanzia l'oggetto per attendere la presenza di dati sulla pipe di output
		this.pOutputPipeWaiter = new Waiter();

		// Inizializza le altre variabili
		this.pLocalAddress			= parLocalAddress;
		this.pChannelInputStream 	= null;
		this.pChannelOutputStream 	= null;
		this.pEventBasedInput 		= false;	
		this.pThreadSender 			= null;
		this.pThreadReceiver 		= null;
		
	}
	
	/**
	 * Costruttore. Inizializza un canale con supporto headerless
	 * @param parName Nome del canale
	 * @param parFrameReceiver Ricevitore frame headerless
	 * @param parFrameSender Oggetto per inviare i frames headerless
	 * @param parLocalAddress Indirizzo locale del canale
	 */
	public Channel(String parName, FrameReceiverHeaderLess parFrameReceiver, 
			FrameSenderHeaderLess parFrameSender, FrameAddress parLocalAddress) {
					
		this(parName, (FrameReceiver) parFrameReceiver, (FrameSender) parFrameSender, parLocalAddress);
		
		// Crea un gestore di eventi per il ricevitore di frame
		this.pFrameReceiverEvents = new FrameReceiverEvents() {

			// Errore
			public void onError(String parErrorMessage) {
				loggerNotify(Type.Warning, "Errore durante ricezione frame", parErrorMessage, Verbosity.Medium);
			}
			
			// Ricevuto un frame headerful
			public void onFrameReceivedHeaderFul(byte[] parFrame, FrameHeader parFrameHeader) {}

			// Ricevuto un frame headerless
			public void onFrameReceivedHeaderLess(byte[] parFrame) {
				
				String tMessage;
				
				tMessage = "Ricevuto frame (";
				for(int iReceivedByte = 0; iReceivedByte < parFrame.length; ++iReceivedByte) {
					if(iReceivedByte > 0) tMessage += ", ";
					tMessage += parFrame[iReceivedByte];
				}
				tMessage += ")";
				loggerNotify(Type.Information, tMessage, Verbosity.High);
				
				pInputPipe.enqueue(new AdapterChannelMessage(getRemoteAddress(), parFrame));			
				
			}
		};
		
		// Istanzia il FrameReceiver
		this.pFrameReceiver.initialize(this.pFrameReceiverEvents);

		// Istanzia l'oggetto per ricevere gli eventi dal FrameSender
		this.pFrameSenderEvents = new FrameSenderEvents() {

			// Errore
			public void onError(String parErrorMessage) {							
				loggerNotify(Type.Warning, "Errore di invio frame", parErrorMessage, Verbosity.Medium);
			}

		};
		
		// Istanzia il FrameSender
		this.pFrameSender.initialize(this.pFrameSenderEvents);

	}
	
	/**
	 * Costruttore. Inizializza un canale con supporto headerful
	 * @param parName Nome canale
	 * @param parFrameReceiver Ricevitore frames headerful
	 * @param parFrameSender Oggetto per inviare i frames headerful
	 * @param parLocalAddress Indirizzo locale nodo
	 */
	public Channel(String parName, FrameReceiverHeaderFul parFrameReceiver, 
			FrameSenderHeaderFul parFrameSender, FrameAddress parLocalAddress) {
		
		this(parName, (FrameReceiver) parFrameReceiver, (FrameSender) parFrameSender, parLocalAddress);
	
		// Crea un gestore di eventi per il ricevitore di frame
		this.pFrameReceiverEvents = new FrameReceiverEvents() {

			// Errore
			public void onError(String parErrorMessage) {
				loggerNotify(Type.Warning, "Errore durante ricezione frame", parErrorMessage, Verbosity.Medium);
			}
			
			// Ricevuto un frame headerful
			public void onFrameReceivedHeaderFul(byte[] parFrame, FrameHeader parFrameHeader) {
				
				String tMessage;
				
				tMessage = "Ricevuto frame :" + parFrameHeader.toString() + " (";
				for(int iReceivedByte = 0; iReceivedByte < parFrame.length; ++iReceivedByte) {
					if(iReceivedByte > 0) tMessage += ", ";
					tMessage += parFrame[iReceivedByte];
				}
				tMessage += ")";

				// Controlla che l'indirizzo destinazione corrisponda a questa macchina
				if ( parFrameHeader.getDestinationAddress().equals(pLocalAddress)) {
					pInputPipe.enqueue(new AdapterChannelMessage(parFrameHeader.getSourceAddress(),parFrame));
					loggerNotify(Type.Information, tMessage, Verbosity.High);
				}
					
				// Il pacchetto era indirizzato a un'altra macchina
				else {
					tMessage += " - indirizzato a un altro nodo.";
					loggerNotify(Type.Information, tMessage, Verbosity.High);
				}
				
			}

			// Ricevuto un frame AddressLess
			public void onFrameReceivedHeaderLess(byte[] parFrame) {}
		};
		
		// Istanzia il FrameReceiver
		this.pFrameReceiver.initialize(this.pFrameReceiverEvents);

		// Istanzia l'oggetto per ricevere gli eventi dal FrameSender
		this.pFrameSenderEvents = new FrameSenderEvents() {

			// Errore
			public void onError(String parErrorMessage) {							
				loggerNotify(Type.Warning, "Errore di invio frame", parErrorMessage, Verbosity.Medium);
			}

		};
		
		// Istanzia il FrameSender
		this.pFrameSender.initialize(this.pFrameSenderEvents);
		
	}

	/**
	 * Imposta il flag che indica se la gestione del canale di input deve essere ad eventi
	 * @param parEventBasedInput
	 */
	protected void setEventBasedInput(boolean parEventBasedInput) {
		
		this.pEventBasedInput = parEventBasedInput;
	}
	
	
	/**
	 * Restituisce il flag che indica se la gestione del canale di input deve essere ad eventi
	 * @return
	 */
	protected boolean getEventBasedInput() {
		
		return this.pEventBasedInput;
	}
	
	
	/**
	 * Imposta il riferimento all'InputStream per leggere dal canale 
	 * @param parInputStream
	 */
	protected void setInputStream(InputStream parInputStream) {
		
		this.pChannelInputStream = parInputStream;
	}
	
	
	/**
	 * Restituisce il riferimento all'InputStream per leggere dal canale
	 * @return
	 */
	protected InputStream getInputStream() {
		return this.pChannelInputStream;
	}
	
	
	/**
	 * Imposta il riferimento all'OutputStream per scrivere sul canale 
	 * @param parOutputStream
	 */
	protected void setOutputStream(OutputStream parOutputStream) {
		this.pChannelOutputStream = parOutputStream;
	}
	
	
	/**
	 * Restituisce il riferimento all'OutputStream per scrivere sul canale
	 * @return
	 */
	protected OutputStream getOutputStream() {
		return this.pChannelOutputStream;
	}
	
	/**
	 * Aggiungi un indirizzo di una macchina raggiungibile attraverso questo canale
	 * @param parRemoteAddress Indirizzo remoto
	 */
	public void setRemoteAddress(FrameAddress parRemoteAddress) {
		this.pRemoteAddress = parRemoteAddress;
	}
	
	/**
	 * Restituisci l'indirizzo della macchina all'altro capo del canale
	 * @return
	 */
	public FrameAddress getRemoteAddress() {
		return this.pRemoteAddress;
	}
	
	/**

	
	/**
	 * Restituisce il riferimento alla Pipe da utilizzare per ricevere dati dal canale
	 * @return
	 */
	public Pipe<AdapterChannelMessage> getInputPipe() {
		return this.pInputPipe;
	}
	
	
	/**
	 * Restituisce il riferimento alla Pipe da utilizzare per inviare dati sul canale
	 */
	public Pipe<AdapterChannelMessage> getOutputPipe() {
		return this.pOutputPipe;
	}
	
	
	/**
	 * Funzione che deve essere overridata dalle classi figlie per aprire
	 * l'input e l'output stream verso il canale.
	 */
	abstract protected void openChannel();
	
	
	/**
	 * Funzione che deve essere overridata dalle classi figlie per chiudere
	 * l'input e l'output stream verso il canale.
	 */
	abstract protected void closeChannel();
	
	
	/**
	 * Avvia il canale, apre l'InputStream e l'OutputStream e predispone i thread necessari.
	 */
	public void start() {
		
		// Verifica che il canale non sia già avviato
		if(!this.isStarted()) {

			// Resetta il frame sender ed il frame receiver
			this.pFrameReceiver.reset();
			this.pFrameSender.reset();
			
			// Avvia le Pipe di input
			this.pInputPipe.loggerDetachAll();
			this.pInputPipe.loggerAttach(this);
			this.pInputPipe.start();
			
			// Prova ad aprire gli Stream verso il canale fisico
			this.openChannel();
			
			// Se gli stream sono stati aperti ..
			if((this.pChannelInputStream != null) && (this.pChannelOutputStream != null)) {
				
				// .. avvia la Pipe di output
				this.pOutputPipe.loggerDetachAll();
				this.pOutputPipe.loggerAttach(this);
				this.pOutputPipe.start();
				
				// .. inizializza e avvia l'Waiter				
				this.pOutputPipeWaiter.addWaitable(this.pOutputPipe.getWaitable());
				this.pOutputPipeWaiter.start();
				
				// .. avvia il thread di scrittura
				this.pThreadSender = new Thread() {
					public void run() {
						pSenderThreadFunction();
					}
				};
				this.pThreadSender.start();
				
				// .. avvia il thread di lettura, se richiesto dal tipo di canale
				if(!this.pEventBasedInput) {
					
					this.pThreadReceiver = new Thread() {
						public void run() {
							pReceiverThreadFunction();
						}
					};
					this.pThreadReceiver.start();
				}
				
				// .. imposta il flag che indica che il componente è stato avviato
				this.setStarted(true);
			}
			else {
				
				this.pInputPipe.stop();			
			}
		}
	}
	
	
	/**
	 * Arresta il canale, chiude l'InputStream e l'OutputStream e termina i thread avviati.
	 */
	public void stop() {
		
		// Verifica che il canale sia avviato
		if(this.isStarted()) {
			
			// Richiede e attende la terminazione del thread di invio
			this.pOutputPipeWaiter.stop();
			try {
				this.pThreadSender.join();
			}
			catch(Exception parException) { }
			this.pThreadSender = null;
			
			// Richiede la terminazione del thread di ricezione, se era stato avviato
			if(!this.pEventBasedInput) {
				this.pThreadReceiver.interrupt();				
			}
			
			// Chiude gli stream verso il canale fisico
			this.closeChannel();
			this.pChannelInputStream = null;
			this.pChannelOutputStream = null;
			
			// Attende la terminazione del Thread di ricezione, se era stato avviato
			if(!this.pEventBasedInput)
			{
				try {					
					this.pThreadReceiver.join();
				}
				catch(Exception parException) { };
				this.pThreadReceiver = null;
			}
			
			// Arresta le pipe
			this.pInputPipe.stop();
			this.pOutputPipe.stop();
		
			// Imposta il flag che indica che il componente è arrestato
			this.setStarted(false);
		}
	}
	
	
	/**
	 * Funzione eseguita dal thread di invio sul canale
	 */
	@SuppressWarnings("unchecked")
	private void pSenderThreadFunction() {
		
		Waitable tWaitable;
		Pipe<AdapterChannelMessage> tPipe;
		AdapterChannelMessage tFrameToSend;
		
		boolean tIOError = false;		// Indica se si è verificato un errore di IO
		String tIOErrorMessage = "";	// Testo dell'eventuale messaggio di errore di IO
		int tIOErrorCount = 0;			// Mantiene un conteggio degli errori di IO consecutivi
		
		
		// Cicla finchè riesce ad ottenere dati da inviare
		do {
			
			// Attende il prossimo pacchetto da inviare
			tWaitable = this.pOutputPipeWaiter.waitNext();			
			if(tWaitable != null) {
				
				// Determina la pipe da cui prelevare il pacchetto da inviare
				tPipe = (Pipe <AdapterChannelMessage>)tWaitable.getParentComponent();
				
				// Estrae dalla pipe il pacchetto da inviare
				tFrameToSend = tPipe.dequeue();
				if(tFrameToSend != null) {
									
					try {
						
						Frame tFrame;
						
						if (this.pFrameSender instanceof FrameSenderHeaderLess) {
							tFrame = new FrameHeaderLess(tFrameToSend.getPayload());
						} else {
							
							FrameHeader tFrameHeader = this.pFrameSender.getConfiguration().getFrameHeader();
							tFrameHeader.setSourceAddress(this.pLocalAddress);
							tFrameHeader.setDestinationAddress(this.pRemoteAddress);
							
							tFrame = new FrameHeaderFul(tFrameHeader, tFrameToSend.getPayload());
						}
					
						this.pChannelOutputStream.write(this.pFrameSender.forgeFrame(tFrame));				
					}
					catch(Exception parException) 
					{
						tIOError = true;
						tIOErrorMessage = parException.getMessage();
						tIOErrorCount ++;											
					}
					
					tFrameToSend = null;
				}
				tPipe = null;
								
				// Se si sono verificati errori sul canale..
				if(tIOError) { 
					
					// .. segnala l'errore verificatosi e attende un timeout prima di riprovare
					tIOErrorCount ++;
					this.loggerNotify(Type.Error, "Problemi di invio sul canale fisico (" + tIOErrorCount + ")", tIOErrorMessage, Verbosity.Low);
					try { 
						Thread.sleep(100);
					}
					catch(Exception parException) { } 
					tIOError = false;
				}
				else {
					tIOErrorCount = 0;
				}							
			}
		
		} while(tWaitable != null);		
	}
	
	
	/**
	 * Funzione eseguita dal Thread di ricezione dal canale
	 */
	private void pReceiverThreadFunction() {
		
		this.pReceive(true);
	}
	
	/**
	 * Preleva i byte presenti nell'InputStream e li fa passare attraverso il FrameReceiver
	 */
	protected final void receive() {
		
		this.pReceive(false);
	}
	
	
	/**
	 * Preleva i byte presenti nell'InputStream e li fa passare attraverso il FrameReceiver
	 * @paam parBLockingMode Indica se deve essere effettuata una lettura bloccante sul canale o se invece devono essere letti i soli byte disponibili
	 */
	private void pReceive(boolean parBlockingMode) {
	
		boolean tHasBytes = true;		// Indica se ci sono byte disponibili nell'inputStream
		int tReadValue = 0;				// Ultimo byte letto dall'input stream (in forma di int)
		byte tReadByte = 0;				// Ultimo byte letto dall'input stream
		
		boolean tIOError = false;		// Indica se si è verificato un errore di IO
		String tIOErrorMessage = "";	// Testo dell'eventuale messaggio di errore di IO
		int tIOErrorCount = 0;			// Mantiene un conteggio degli errori di IO consecutivi
		boolean tTerminate = false;		// Indica se il Thread deve essere terminato
		
		
		// Cicla finchè ci sono byte disponibili nell'InputStream..		
		do {
			
			// Determina se ci sono byte disponibili
			if(!parBlockingMode) {
				try {
					tHasBytes = (this.pChannelInputStream.available() > 0);
				}
				catch(IOException parException) {
					tIOError = true;
					tIOErrorMessage = parException.getMessage();					
					tHasBytes = false;
				}
			}
			else
			{
				tHasBytes = true;
			}
			
			// Legge il prossimo byte
			if(tHasBytes) {
				try {
					tReadValue = this.pChannelInputStream.read();
					if (tReadValue == -1) {
						tTerminate = true;
						tHasBytes = false;
					}
				}
				catch(IOException parException) {	
					tIOError = true;
					tIOErrorMessage = parException.getMessage();					
					tHasBytes = false;
				}
			}
			
			// Passa il byte letto attraverso il FrameReceiver
			if(tHasBytes) {											
				tReadByte = (byte)tReadValue;
				this.pFrameReceiver.receiveByte(tReadByte);
			}
			
			// Se si sono verificati errori sul canale..
			if(tIOError) {
				
				//  .. controlla se è stata richiesta una terminazione del Thread
				if(Thread.interrupted())
				{
					tTerminate = true;
					System.out.println(tIOErrorMessage);
				}
				
				// .. segnala l'errore verificatosi e attende un timeout prima di riprovare
				else {
					tIOErrorCount ++;
					this.loggerNotify(Type.Error, "Problemi di ricezione sul canale fisico (" + tIOErrorCount + ")", tIOErrorMessage, Verbosity.Low);
					try {
						Thread.sleep(100);
					}
					catch(Exception parException) { } 
				}
				tIOError = false;
			}
			else {
				tIOErrorCount = 0;
			}
			
		} while((!tTerminate) && (tHasBytes || parBlockingMode));	
		
	}
	
}
